package com.hancheng.commoncloud.framework.util;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hancheng.commoncloud.framework.result.exception.CommonException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class KafkaSenderUtils {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    public KafkaSenderUtils(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        this.objectMapper = new ObjectMapper();
    }

    public void send(String topic, Object message) {
        send(topic, null, message);
    }

    public void send(String topic, String key, Object message) {
        try {
            String jsonMessage = objectMapper.writeValueAsString(message);
            if (key != null) {
                kafkaTemplate.send(topic, key, jsonMessage);
            } else {
                kafkaTemplate.send(topic, jsonMessage);
            }
            log.info("Kafka发送消息 => topic: {}, key: {}, message: {}", topic, key, jsonMessage);
        } catch (JsonProcessingException e) {
            throw new CommonException(e.getMessage());
        }
    }
}
